package org.zn.note.java8;

import org.zn.note.java8.Bean.Dish;

import java.util.List;
import java.util.stream.IntStream;
import java.util.stream.Stream;

/**
 * 并行数据处理与性能
 * <p>
 * Stream是可以切换线程是单线程执行(串行)、还是多线程执行(并行)，但是效果不是完全可控的
 * 例子 {@link #testParallelAndSequential()}
 * 结论：要么就都用单线程，要么都切换到多线程
 * <p>
 * 并行的线程从哪来？
 * 并行线程默认是ForkJoinPool，默认线程数是处理器数量，Runtime.getRuntime().available-Processors()
 * 可以改 System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism","12");
 * 但没必要
 * <p>
 * 一些主要注意的：
 * 1、改为并行，性能是否提高，是否线程安全，都要测试
 * 2、自动装箱、拆箱，为大大降低性能，这个要注意，如果是基础数据类型，尽量用IntStream、LongStream、DoubleStream做
 * 3、有些操作本身是带有顺序性的，如果改成并行的，就是会慢。比如
 * {@link Stream#limit(long)}，因为流本身是带有顺序的数据，所以这个的含义是“前N个元素”
 * {@link Stream#findFirst()}，找出第一个 （如果可能，尽量有{@link Stream#findAny()}）
 * 可以调用{@link Stream#unordered()}，把流变成无序的，就没事了
 * 4、数据结构本身是否易于分解：
 * {@link java.util.ArrayList} 极佳
 * {@link java.util.LinkedList} 差
 * {@link IntStream#range(int, int)} 极佳
 * {@link Stream#iterator()} 差
 * {@link java.util.HashSet} 还行
 * {@link java.util.TreeSet} 还行
 * 5、终端操作，合并流的代价
 * <p>
 * Spliterator
 * Stream的并行是通过jdk1.7的fork/join框架做的，参考 {@link Ch07_ForkJoin}
 * 那么Stream是如何拆分子任务的呢？ 用Spliterator (jdk1.8 可分迭代器 splitable iterator)
 * 和Iterator一样，也是用于遍历数据源的元素的，只不过这是“可拆分”的！
 */
public class Ch07_Parallel {

    private static final List<Dish> menu = Dish.genDishs();

    public static void main(String[] args) {
        testParallelAndSequential();
    }

    // 生成N个自然数的和
    private static long sequentialSum(long n) {
        return Stream
                .iterate(1L, i -> i + 1)
                .limit(n)
                .reduce(0L, Long::sum);
    }

    // 并行方案
    private static long parallelSum(long n) {
        return Stream.iterate(1L, i -> i + 1)
                .limit(n)
                .parallel()
                .reduce(0L, Long::sum);
    }

    /**
     * 看下并行和串行的执行效果：
     * 刚开始任务1、任务2、任务3都是单线程的，线程名：main
     * 后来又全是多线程执行了，线程名：ForkJoinPool.commonPool-worker-1/2/3
     */
    private static void testParallelAndSequential() {
        IntStream.range(1, 100).boxed()
                .parallel().filter(i -> {
            System.out.printf("任务1：[%d] [%s]\n", i, Thread.currentThread().getName());
            return true;
        })
                .sequential().filter(i -> {
            System.out.printf("任务2：[%d] [%s]\n", i, Thread.currentThread().getName());
            return true;
        })
                .parallel().filter(i -> {
            System.out.printf("任务3：[%d] [%s]\n", i, Thread.currentThread().getName());
            return true;
        })
                .count();
    }

}